use std::{collections::HashMap, fs, path::PathBuf, sync::Arc};
use log::warn;
use parking_lot::RwLock;

use crate::{
    data::{
        self, data_file::DataFile, log_record::{self, LogRecord, LogRecordPos, LogRecordType}
    },
    errors::{self, Errors, Result},
    index,
    options::options,
};

pub struct Engine {
    options: Arc<options>,
    active_file: Arc<RwLock<DataFile>>,
    older_files: Arc<RwLock<HashMap<u32, DataFile>>>,
    pub(crate) index: Box<dyn index::Indexer>,
    file_ids:Vec<u32>,
}

impl Engine {

    pub fn open(opts:options) -> Result<Self> {
        // 配置项进行校验
        if let Some(e) = check_options(&opts) {
            return Err(e);
        }
        // 判断数据目录是否存在，不存在则进行创建
        let dir_path = opts.dir_path.clone();
        if !dir_path.is_dir() {
            if let Err(e) = fs::create_dir_all(dir_path.clone()){
                warn!("create database dir err:{}",e);
                return Err(errors::Errors::FailedCreateDataDirError);
            }
        }

        // 加载数据文件
        let mut data_files = load_data_files(dir_path.clone())?;

        // 设置file_id信息
        let mut file_ids = Vec::new();
        for data_file in data_files.iter(){
            file_ids.push(data_file.get_file_id());
        }
        // 
        data_files.reverse();
        // 将旧的数据文件保存到older_files中
        let mut older_files = HashMap::new();
        if data_files.len() > 1 {
            for _ in 0..=data_files.len()-2 {
                let file = data_files.pop().unwrap();
                older_files.insert(file.get_file_id(), file);
            }
        }

        // 拿到当前活跃文件
        let active_file = match data_files.pop() {
            Some(v) => v,
            None => DataFile::new(dir_path.clone(), 0)?,
        };

        // 构造数据库引擎实例
        let mut engine = Engine {
            options:Arc::new(opts.clone()),
            active_file:Arc::new(RwLock::new(active_file)),
            older_files:Arc::new(RwLock::new(older_files)),
            index: Box::new(index::new_indexer(opts.index_type)),
            file_ids,
        };

        // 从数据文件加载索引
        engine.load_index_from_data_files();

        Ok(engine)
    }

    pub fn put(&self, key: Vec<u8>, value: Vec<u8>) -> Result<()> {
        if key.is_empty() {
            return Err(crate::errors::Errors::KeyIsEmptyError);
        }

        let mut record = LogRecord {
            key: key.clone(),
            value: value,
            rec_type: LogRecordType::NORMAL,
        };

        let log_record_pos = self.append_log_record(&mut record)?;

        //println!("log_record_pos: {:?}",log_record_pos);

        let ok = self.index.put(key, log_record_pos);

        if ok != true {
            return Err(errors::Errors::IndexUpdateError);
        }

        Ok(())
    }

    pub fn get(&self, key: Vec<u8>) -> Result<Vec<u8>> {
        if key.is_empty() {
            return Err(errors::Errors::KeyIsEmptyError);
        }

        let pos = self.index.get(key);
        if pos.is_none() {
            return Err(errors::Errors::KeyIsNotFound);
        }
        let log_record_pos = pos.unwrap();
        let active_file = self.active_file.read();
        let older_files = self.older_files.read();
        let log_record = match active_file.get_file_id() == log_record_pos.file_id {
            true => active_file.read_log_record(log_record_pos.offset)?,
            false => {
                let data_file = older_files.get(&log_record_pos.file_id);
                if data_file.is_none() {
                    return Err(errors::Errors::DataFileNotFound);
                }
                data_file.unwrap().read_log_record(log_record_pos.offset)?
            }
        };
        if log_record.record.rec_type == LogRecordType::DELETED {
            return Err(errors::Errors::KeyIsNotFound);
        }

        Ok(log_record.record.value)
    }

    pub fn delete(&self, key: Vec<u8>)-> Result<()> {
        // 判断key有效性
        if key.is_empty(){
            return Err(Errors::KeyIsEmptyError);
        }

        // 从内存索引当中取出对应数据，不存在的话直接返回
        let pos = self.index.get(key.clone());
        if pos.is_none() {
            return Ok(());
        }

        // 设置索引类型为删除状态
        let mut record = LogRecord {
            key:key.clone(),
            value: Default::default(),
            rec_type: LogRecordType::DELETED,
        };

        // 写入数据文件中
        self.append_log_record(&mut record)?;

        let ok = self.index.delete(key.clone());
        if !ok {
            return Err(errors::Errors::IndexUpdateError);
        }

        Ok(())
    }
    
    // 写数据到数据文件
    fn append_log_record(&self, record: &mut LogRecord) -> Result<LogRecordPos> {
        let dir_path = self.options.dir_path.clone();

        let enc_record = record.encode();
        //println!("enc_record:{:?}",enc_record);
        let record_len = enc_record.len() as u64;

        let mut active_file = self.active_file.write();

        if active_file.get_write_off() + record_len > self.options.data_file_size {
            active_file.sync()?;

            let current_fid = active_file.get_file_id();

            let mut older_files = self.older_files.write();
            let old_file = DataFile::new(dir_path.clone(), current_fid)?;
            older_files.insert(current_fid, old_file);

            let new_file = DataFile::new(dir_path.clone(), current_fid + 1)?;

            *active_file = new_file;
        }

        let write_off = active_file.get_write_off();
        active_file.write(&enc_record)?;

        if self.options.sync_writes {
            active_file.sync()?;
        }
        //println!("write_off:{},len:{}",write_off,enc_record.len());
        Ok(LogRecordPos {
            file_id: active_file.get_file_id(),
            offset: write_off,
        })
    }

    //从数据文件中加载内存索引
    fn load_index_from_data_files(&mut self) -> Result<()> {
        // 数据文件为空直接返回
        if self.file_ids.is_empty(){
            return Ok(());
        }
        let mut active_file = self.active_file.read();
        let mut older_files = self.older_files.read();

        // 遍历每个文件，加载数据
        for file_id in self.file_ids.iter() {
            let mut offset = 0u64;
            loop {
                let log_record_res = match *file_id == active_file.get_file_id() {
                    true => {
                        active_file.read_log_record(offset)
                    },
                    false => {
                        older_files.get(file_id).unwrap().read_log_record(offset)
                    }
                };
                let (log_record,size) = match log_record_res {
                    Ok(res) => (res.record,res.size),
                    Err(e) => {
                        if e == Errors::ReadDataEOFError {
                            break;
                        }
                        return Err(e);
                    }
                };


                let log_record_pos = LogRecordPos {
                    file_id: *file_id,
                    offset,
                };

                match log_record.rec_type {
                    LogRecordType::NORMAL => {
                        self.index.put(log_record.key,log_record_pos);
                    }
                    LogRecordType::DELETED => {
                        self.index.delete(log_record.key);
                    }
                }

                offset += size;
            }
            if *file_id == active_file.get_file_id() {
                active_file.set_write_off(offset);
            }
        }

        Ok(())
    }

}

fn load_data_files(dir_path:PathBuf) -> Result<Vec<DataFile>> {
    // 读取数据目录
    let dir = fs::read_dir(dir_path.clone());
    if dir.is_err() {
        return Err(Errors::FailedReadDataDirError);
    }
    let mut file_ids = Vec::new();
    let mut data_files = Vec::new();
    for file in dir.unwrap() {
        if let Ok(entry) = file {
            //拿到文件名
            let file_os_str = entry.file_name();
            let file_name = file_os_str.to_str().unwrap();

            //判断文件名是不是.data结尾
            if file_name.ends_with(data::data_file::DATA_FILE_NAME_SUFFIX){
                let split_names:Vec<&str> = file_name.split(".").collect();
                let file_id = match split_names[0].parse::<u32>() {
                    Ok(fid) => fid,
                    Err(_) => {
                        return Err(Errors::DataDirCorrupted);
                    }
                };
                file_ids.push(file_id);
            }
        }

    }

    if file_ids.is_empty(){
        return Ok(data_files);
    }

    file_ids.sort();

    for file_id in file_ids.iter() {
        let data_file = DataFile::new(dir_path.clone(), *file_id)?;
        data_files.push(data_file);
    }

    Ok(data_files)
}

fn check_options(opts:& options) -> Option<Errors> {
    let dir_path = opts.dir_path.to_str();
    if dir_path.is_none() || dir_path.unwrap().len() == 0 {
        return Some(Errors::DirPathIsEmpty);
    }
    if opts.data_file_size <= 0 {
        return Some(Errors::DataFileSizeTooSmall);
    }
    None
}